package com.atguigu.edu.realtime.app.dwd.trade;

import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author maikle
 * @date 2022/8/2
 * 交易域：订单预处理表
 * 需要启动的进程
 *      zk、kafka、maxwell、DwdTradeOrderPreProcess
 */
public class DwdTradeOrderPreProcess {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(3);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15*60 + 10));

        //TODO 2.检查点相关的设置(略)

        //TODO 3.从kafka的topic_db主题中读取数据 创建对应的动态表
        tableEnv.executeSql("create table ods_db(" +
            "`database` String,\n" +
            "`table` String,\n" +
            "`type` String,\n" +
            "`data` map<String, String>,\n" +
            "`old` map<String, String>,\n" +
            "`proc_time` as PROCTIME(),\n" +
            "`ts` string\n" +
            ")" + KafkaUtil.getKafkaDDL("ods_db", "dwd_trade_order_pre_process"));

        //tableEnv.executeSql("select * from ods_db").print();
        //TODO 4.过滤出订单明细
        Table orderDetail = tableEnv.sqlQuery("select \n" +
            "data['id'] id,\n" +
            "data['course_id'] course_id,\n" +
            "data['course_name'] course_name,\n" +
            "data['order_id'] order_id,\n" +
            "data['user_id'] user_id,\n" +
            "data['create_time'] create_time,\n" +
            "data['origin_amount'] origin_amount,\n" +
            "data['coupon_reduce'] coupon_reduce,\n" +
            "data['final_amount'] final_amount,\n" +
            "ts od_ts,\n" +
            "proc_time\n" +
            "from `ods_db` where `table` = 'order_detail' " +
            "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_detail", orderDetail);

        //tableEnv.executeSql("select * from order_detail").print();
        //TODO 5.过滤出订单数据
        Table orderInfo = tableEnv.sqlQuery("select \n" +
            "data['id'] id,\n" +
            "data['user_id'] user_id,\n" +
            "data['province_id'] province_id,\n" +
            "data['expire_time'] expire_time,\n" +
            "data['order_status'] order_status,\n" +
            "data['session_id'] session_id,\n" +
            "`type`,\n" +
            "`old`,\n" +
            "ts oi_ts\n" +
            "from `ods_db`\n" +
            "where `table` = 'order_info'\n" +
            "and (`type` = 'insert' or `type` = 'update')");
        tableEnv.createTemporaryView("order_info", orderInfo);
        //tableEnv.executeSql("select * from order_info").print();

        // todo 6. 从mysql业务数据库中读取课程信息表
        // todo 7. 从mysql业务数据库中读取科目表
        // todo 8. 从mysql业务数据库中读取分类表
        //TODO 6.对上述2张表进行连接
        Table resultTable = tableEnv.sqlQuery("select \n" +
            "od.id,\n" +
            "od.order_id,\n" +
            "od.course_id,\n" +
            "od.course_name,\n" +
            "date_format(od.create_time, 'yyyy-MM-dd') date_id,\n" +
            "od.create_time,\n" +


            "od.origin_amount,\n" +
            "od.coupon_reduce,\n" +
            "od.final_amount,\n" +
            "od.od_ts,\n" +

            "oi.user_id,\n" +
            "oi.province_id,\n" +
            "date_format(oi.expire_time, 'yyyy-MM-dd') expire_time_date,\n" +
            "oi.expire_time,\n" +
            "oi.session_id,\n" +
            "oi.`type`,\n" +
            "oi.`old`,\n" +
            "oi.oi_ts,\n" +

            "current_row_timestamp() row_op_ts\n" +
            "from order_detail od \n" +
            "join order_info oi\n" +
            "on od.order_id = oi.id\n");
        tableEnv.createTemporaryView("result_table", resultTable);
        //tableEnv.executeSql("select * from result_table").print();
        //TODO 7.创建表和要写出kafka主题进行映射
        tableEnv.executeSql("" +
            "create table dwd_trade_order_pre_process(\n" +
            "id string,\n" +
            "order_id string,\n" +
            "course_id string,\n" +
            "course_name string,\n" +
            "date_id string,\n" +
            "create_time string,\n" +

            "origin_amount string,\n" +
            "coupon_reduce string,\n" +
            "final_amount string,\n" +
            "od_ts string,\n" +

            "user_id string,\n" +
            "province_id string,\n" +
            "expire_time_date string,\n" +
            "expire_time string,\n" +
            "session_id string,\n" +
            "`type` string,\n" +
            "`old` map<string,string>,\n" +
            "oi_ts string,\n" +

            "row_op_ts timestamp_ltz(3),\n" +
            "primary key(id) not enforced\n" +
            ")" + KafkaUtil.getUpsertKafkaDDL("dwd_trade_order_pre_process"));


        //TODO 8.将关联的结果写到kafka对应的主题中
        tableEnv.executeSql("insert into dwd_trade_order_pre_process \n" +
            "select * from result_table");


    }
}
